/**
 * CPAR TOOLKIT - Compiler and Runtime for CPAR
 * Copyright (C) 2012 CPAR TOOLKIT Project Developers
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <windows.h>
#include "cpar.h"

typedef struct CPARWorkPoolThread     CPARWorkPoolThread;
typedef struct CPARWorkPoolWorkEntry  CPARWorkPoolWorkEntry;

struct CPARWorkPoolWorkEntry
{
  CPARWorkPoolWork          work;
  void                      *data;
  CPARWorkPoolWorkListener  *listener;
};

struct CPARWorkPoolThread
{
  HANDLE        handle;
  CPARWorkPool  *pool;
};

struct CPARWorkPool
{
  CPARPointerArray    *threads;
  cpar_atomic         activeCount;
  cpar_atomic         runningCount;
  CPAREvent           *shutdown;
  CPAREvent           *closed;
  CPARAllocator       *allocator;
  long                allocatorListenerId;
  CPARConcurrentQueue *queue;
  CPARMutex           *lock;
};

static cpar_status _CPARWorkPoolThreadCreate( CPARWorkPool *workPool );
static void _CPARWorkPoolAllocatorListener(CPARAllocator *allocator, void *data);
static DWORD _CPARWorkPoolThreadRun(void *data);

static
void
_CPARWorkPoolAllocatorListener(CPARAllocator *allocator, void *data)
{
  CPARWorkPool  *pool;
  
  pool = (CPARWorkPool*)data;

  CPARWorkPoolDestroy(pool);
}

static
DWORD 
_CPARWorkPoolThreadRun(void *data)
{
  CPARWorkPoolThread    *pThread;
  CPARWorkPool          *pPool;
  CPARWorkPoolWorkEntry *pWork;
  cpar_atomic           cnt;

  pThread = (CPARWorkPoolThread*)data;
  pPool   = pThread->pool;

  CPARAtomicInc( &(pThread->pool->activeCount) );

  while(TRUE)
  {
    if(CPAREventIsSet(pPool->shutdown))  
      break;

    pWork = (CPARWorkPoolWorkEntry*) CPARConcurrentQueueTryDequeue(pThread->pool->queue, 1000);
    if(NULL == pWork)
      continue;

    cnt = CPARAtomicInc(&(pPool->runningCount));

    if(pWork->listener && pWork->listener->onBeforeRun)
      pWork->listener->onBeforeRun(pWork->data);

    pWork->work(pWork->data);

    if(pWork->listener && pWork->listener->onAfterRun)
      pWork->listener->onAfterRun(pWork->data);
    
    cnt = CPARAtomicDec(&(pPool->runningCount));

    CPARAllocatorFree(pThread->pool->allocator,pWork);
  }

  CloseHandle( pThread->handle );

  cnt = CPARAtomicDec( &(pThread->pool->activeCount) );

  if(cnt == 0)
    CPAREventSet(pPool->closed);

  return 0L;
}

static
cpar_status
_CPARWorkPoolThreadCreate( CPARWorkPool *workPool )
{
  CPARWorkPoolThread  *pThread;
  cpar_status         s;
  long                idx;

  pThread = (CPARWorkPoolThread*) CPARAllocatorAlloc(workPool->allocator,sizeof(CPARWorkPoolThread));
  if(NULL == pThread)
    return CPAR_E_NO_MEMORY;

  CPARMutexLock(workPool->lock);
  s = CPARPointerArrayAdd(workPool->threads, pThread, &idx);
  CPARMutexUnlock(workPool->lock);

  if(s)
    return s;

  pThread->pool   = workPool;
  pThread->handle = 0;

  pThread->handle = CreateThread(NULL, 1024*1024, _CPARWorkPoolThreadRun, pThread, 0, NULL);
  if(INVALID_HANDLE_VALUE == pThread->handle)
  {
    CPARMutexLock(workPool->lock);
    CPARPointerArrayRemove(workPool->threads, idx);
    CPARMutexUnlock(workPool->lock);
    return CPAR_E_GENERAL;
  }

  return CPAR_SUCCESS;
}

cpar_status
CPARWorkPoolCreate( CPARAllocator *allocator, CPARWorkPool **workpool, long maxWorkers )
{
  CPARWorkPool  *p;
  cpar_status   s;

  p = (CPARWorkPool*) CPARAllocatorAlloc(allocator,sizeof(CPARWorkPool));
  if(NULL == p)
    return CPAR_E_NO_MEMORY;

  p->threads      = NULL;
  p->activeCount  = 0;
  p->runningCount = 0;
  p->allocator    = allocator;
  p->shutdown     = NULL;
  p->queue        = NULL;
  p->allocatorListenerId = CPARAllocatorDestroyListenerAdd( allocator, _CPARWorkPoolAllocatorListener, p );

  s = CPARMutexCreate(allocator, &(p->lock));
  if(s)
  {
    CPARWorkPoolDestroy(p);
    return s;
  }

  s = CPARPointerArrayCreate(allocator, &(p->threads), maxWorkers, 5);
  if(s)
  {
    CPARWorkPoolDestroy(p);
    return s;
  }

  s = CPAREventCreate(allocator, &(p->shutdown));
  if(s)
  {
    CPARWorkPoolDestroy(p);
    return s;
  }

  s = CPAREventCreate(allocator, &(p->closed));
  if(s)
  {
    CPARWorkPoolDestroy(p);
    return s;
  }
  
  s = CPARConcurrentQueueCreate(allocator,&(p->queue));
  if(s)
  {
    CPARWorkPoolDestroy(p);
    return s;
  }

  *workpool = p;

  return CPARWorkPoolSetMaxWorkers(p,maxWorkers);
}

void 
CPARWorkPoolDestroy( CPARWorkPool *workPool )
{
  if(workPool->shutdown)
  {
    CPAREventSet(workPool->shutdown);
    
    if(workPool->closed)
    {
      CPAREventWait(workPool->closed);
      CPAREventDestroy(workPool->closed);
    }

    CPAREventDestroy(workPool->shutdown);
  }

  if(workPool->queue)
    CPARConcurrentQueueDestroy(workPool->queue);

  if(workPool->threads)
    CPARPointerArrayDestroy(workPool->threads);

  if(workPool->lock)
    CPARMutexDestroy(workPool->lock);

  CPARAllocatorFree(workPool->allocator, workPool->threads);
  CPARAllocatorFree(workPool->allocator, workPool);

  CPARAllocatorDestroyListenerRemove(workPool->allocator, workPool->allocatorListenerId);
}

cpar_status
CPARWorkPoolDispatch( CPARWorkPool *workPool, CPARWorkPoolWork work, void *data, CPARWorkPoolWorkListener *listener)
{
  cpar_status s;

  //TODO Memory Pool for works
  CPARWorkPoolWorkEntry *pWork;

  pWork = (CPARWorkPoolWorkEntry*) CPARAllocatorAlloc(workPool->allocator,sizeof(CPARWorkPoolWorkEntry));
  if(NULL == pWork)
    return CPAR_E_NO_MEMORY;

  pWork->work = work;
  pWork->data = data;
  pWork->listener = listener;

  s = CPARConcurrentQueueEnqueue(workPool->queue, pWork);
  if(s != CPAR_SUCCESS)
    CPARAllocatorFree(workPool->allocator, pWork);

  return s;
}

CPARWorkPoolWorkListener*
CPARWorkPoolWorkListenerCreate( CPARAllocator *allocator )
{
  CPARWorkPoolWorkListener *pListener;

  pListener = (CPARWorkPoolWorkListener*)CPARAllocatorAlloc(allocator, sizeof(CPARWorkPoolWorkListener));
  if(NULL == pListener)
    return NULL;

  pListener->allocator    = allocator;
  pListener->onAfterRun   = NULL;
  pListener->onBeforeRun  = NULL;

  return pListener;
}

void 
CPARWorkPoolWorkListenerDestroy( CPARWorkPoolWorkListener *listener )
{
  CPARAllocatorFree(listener->allocator,listener);
}

cpar_status 
CPARWorkPoolSetMaxWorkers( CPARWorkPool *workPool, long maxWorkers )
{
  long toCreate;

  CPARMutexLock(workPool->lock);
  toCreate = CPARPointerArrayCount(workPool->threads);
  CPARMutexUnlock(workPool->lock);
  toCreate = maxWorkers - toCreate;

  for(;toCreate > 0; toCreate--)
  {
    cpar_status s = _CPARWorkPoolThreadCreate(workPool);
    if(s)
      return s;
  }

  return CPAR_SUCCESS;
}